package cn.z2huo.demo.kafka.apache.configtest.producer.batch;

import cn.z2huo.demo.kafka.apache.configtest.producer.DataGeneratorUtils;
import cn.z2huo.demo.kafka.apache.jackson.KafkaJsonMapper;
import cn.z2huo.demo.kafka.apache.producer.AsyncSendMessageCallback;
import cn.z2huo.demo.kafka.apache.producer.ProducerProperties;
import cn.z2huo.demo.model.dataobject.user.UserDO;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.json.JsonMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * <p>
 *
 * @author z2huo
 */
@Slf4j
public class BatchSizeLingerProducer {

    private static final String TOPIC = "test-topic";

    /**
     * 生成的用户数量
     */
    private static final int GENERATE_USER_COUNT = 100;

    public static void main(String[] args) {
        new BatchSizeLingerProducer().send(GENERATE_USER_COUNT);
    }

    public void send(int count) {

        long start, end;

        Properties properties = ProducerProperties.getProducerProperties();
        // 指示生产者等待多达该毫秒数再发送请求，希望有更多记录到达以填充同一个批次
        // 等待批次被填充 10 秒
        properties.put(ProducerConfig.LINGER_MS_CONFIG, "10000");
        // 当多个记录被发送到同一个分区时，生产者会尝试将这些记录批量打包到更少的请求中，此配置用来指示生产者一个批次可以使用的内存大小
        // 配置选择使用 30000，一个用户的 json 字符串大约为 256 个字节，100 个用户就是 25600 字节，设定为 30000 时，100 个用户达不到批次大小，不会立刻发送
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, "30000");

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(properties)) {

            List<UserDO> userDOS = DataGeneratorUtils.generateUser(count);
            JsonMapper jsonMapper = KafkaJsonMapper.INSTANCE;

            start = System.currentTimeMillis();

            long totalBytes = 0L;
            for (int i = 0; i < userDOS.size(); i++) {
                log.info("send the NO.{} user", i);
                String userDOJsonString = jsonMapper.writeValueAsString(userDOS.get(i));
                int bytesLength = userDOJsonString.getBytes(StandardCharsets.UTF_8).length;
                totalBytes += bytesLength;
                log.info("userDO json String length is {}, size is {} bytes, userDOJsonString is {}", userDOJsonString.length(),
                        bytesLength, userDOJsonString);
                ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, userDOJsonString);
                producer.send(record, new AsyncSendMessageCallback());
            }
            // 刷新并关闭生产者
//            producer.flush();

            /*
                主线程 sleep 20 秒，使主线程不关闭
                可以看到，100 个用户，调用完成最后一个 send 方法，到生产者接收到第一个发送成功的回执，时间为 10 秒，为 linger.ms 的配置。
                [11:35:15.968][INFO ] cn.z2huo.demo.kafka.apache.configtest.producer.batch.BatchSizeLingerProducer:63 send - userDO json String length is 258, size is 258 bytes, userDOJsonString is {"id":53258318545862755,"userCode":"userCode kBKaPGlt","userName":"userName xyNopBMJ","validDate":"2025-04-28","invalidDate":"2026-04-28","validFlag":"1","deleteFlag":"0","createTime":"2025-04-28T11:35:15.6422528","operateTime":"2025-04-28T11:35:15.6422528"}
                [11:35:25.953][INFO ] cn.z2huo.demo.kafka.apache.producer.AsyncSendMessageCallback:18 onCompletion - send message to topic test-topic, partition 2 with offset 325
             */
            TimeUnit.SECONDS.sleep(20);

            end = System.currentTimeMillis();
            log.info("send {} users cost {} ms, all json string bytes size is {}", count, end - start, totalBytes);
        } catch (JsonProcessingException e) {
            log.error("write user to json fail");
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

}
